package io.kiki.sba.registry.server.shared.providedata;

import io.kiki.sba.registry.common.model.metaserver.FetchSystemPropertyResult;
import io.kiki.sba.registry.common.model.metaserver.ProvideData;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import lombok.Getter;
import lombok.NonNull;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;


public abstract class AbstractFetchSystemPropertyService<T extends SystemDataStorage> implements FetchSystemPropertyService {

    protected static final long INIT_VERSION = -1L;
    private static final Logger logger = LoggerFactory.getLogger(FetchSystemPropertyService.class);
    @Getter
    protected final AtomicReference<T> storage = new AtomicReference<>();
    protected final WatchDog watchDog = new WatchDog();
    private final String dataInfoId;
    private final AtomicBoolean watcherInited = new AtomicBoolean(false);
    @Autowired
    protected MetaServerService metaNodeService;

    protected AbstractFetchSystemPropertyService(String dataInfoId, @NonNull T t) {

        this.dataInfoId = dataInfoId;
        storage.set(t);
    }

    protected abstract int getSystemPropertyIntervalMillis();

    protected boolean doFetchData() {
        T expect = storage.get();
        FetchSystemPropertyResult response = metaNodeService.fetchSystemProperty(dataInfoId, expect.version);

        StringUtil.nullValue(response, "fetchSystemPropertyResult");

        if (!response.isVersionUpgrade()) {
            return true;
        }

        if (logger.isInfoEnabled()) {
            logger.info("[FetchSystemProperty]dataInfoId:{}, version:{}, response:{}", dataInfoId, expect, response);
        }
        // do process
        return processorData(response.getProvideData(), expect);
    }

    @Override
    public boolean doFetch() {
        watchDog.wakeup();
        return true;
    }

    @Override
    public boolean start() {
        doFetchData();
        if (watcherInited.compareAndSet(false, true)) {
            ConcurrentUtils.createDaemonThread(StringFormatter.format("FetchSystemProperty-{}", dataInfoId), watchDog).start();
        }

        return watcherInited.get();
    }

    private boolean processorData(ProvideData provideData, T expect) {
        Assert.isTrue(provideData != null, StringFormatter.format("[FetchSystemProperty]dataInfoId:{}, versionUpgrade:{}, but provideData is null.", dataInfoId, true));

        try {
            if (provideData.getVersion() <= expect.version) {
                logger.warn("Fetch system data={}, currentVersion={}, updateVersion={}", dataInfoId, expect, provideData.getVersion());
                return false;
            }
        } catch (Throwable e) {
            logger.error("Fetch session stopPushSwitch error.", e);
            return false;
        }

        // do process
        return doProcess(expect, provideData);
    }

    @Override
    public boolean support(String dataInfoId) {
        return StringUtils.equals(this.dataInfoId, dataInfoId);
    }

    protected boolean compareAndSet(T expect, T update) {
        return storage.compareAndSet(expect, update);
    }

    protected abstract boolean doProcess(T expect, ProvideData data);


    protected final class WatchDog extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            doFetchData();
        }

        @Override
        public int getWaitingMillis() {
            return getSystemPropertyIntervalMillis();
        }
    }
}
